package com.rem.flink.flink6ProcessFunction;

import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * ProcessFunction 最基本的处理函数，基于 DataStream 直接调用.process()时作为参数传入。
 * KeyedProcessFunction 对流按键分区后的处理函数，基于 KeyedStream 调用.process()时作为参数传入。要想使用定时器，比如基于 KeyedStream。
 * ProcessWindowFunction 开窗之后的处理函数，也是全窗口函数的代表。基于 WindowedStream 调用.process()时作为参数传入。
 * ProcessAllWindowFunction 同样是开窗之后的处理函数，基于 AllWindowedStream 调用.process()时作为参数传入。
 * CoProcessFunction 合并（connect）两条流之后的处理函数，基于 ConnectedStreams 调用.process()时作为参数传入。
 * ProcessJoinFunction 间隔连接（interval join）两条流之后的处理函数，基于 IntervalJoined 调用.process()时作为参数传入
 * BroadcastProcessFunction 广播连接流处理函数，基于 BroadcastConnectedStream 调用.process()时作为参数传入。是一个未keyBy 的普通 DataStream 与一个广播流（BroadcastStream）做连接（conncet）之后的产物
 * KeyedBroadcastProcessFunction 按键分区的广播连接流处理函数，同样是基于 BroadcastConnectedStream 调用.process()时作为参数传入。与 BroadcastProcessFunction 不同的是，这时的广播连接流，是一个 KeyedStream与广播流（BroadcastStream）做连接之后的产物。
 * <p>
 * <p>
 * KeyedProcessFunctionTest demo 定时器
 *
 * @author Rem
 * @date 2022-10-13
 */

public class KeyedProcessFunctionTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new CustomSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()));

        /**
         * 要用定时器，必须基于 KeyedStream
         */
        stream.keyBy(e -> true)
                .process(new KeyedProcessFunction<Boolean, Event, String>() {


                    @Override
                    public void processElement(Event value, KeyedProcessFunction<Boolean, Event, String>.Context ctx, Collector<String> out) throws Exception {
                        out.collect("数据到达，时间戳为：" + ctx.timestamp());
                        out.collect("数据到达，水位线为：" + ctx.timerService().currentWatermark() + "\n -------分割线-------");
                        // 注册一个10秒后的定时器
                        ctx.timerService().registerProcessingTimeTimer(ctx.timestamp() + 10 * 1000L);
                    }

                    /**
                     * 触发定时器
                     * @param timestamp
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<Boolean, Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect("定时器触发，触发时间：" + new Timestamp(timestamp));
                    }
                }).print();

        env.execute();

    }


    /**
     * 自定义测试数据源
     * 数据第一次到达时间为  Mary 时间戳为2s 此时水位线为 -9223372036854775808 水位线最小值
     * 第二次数据来时 水位线为前一个时间戳减一 为1999
     *
     */
    public static class CustomSource implements SourceFunction<Event> {
        @Override
        public void run(SourceContext<Event> ctx) throws Exception {
            // 直接发出测试数据
            ctx.collect(new Event("Mary", "./home", 1000L));
            // 为了更加明显，中间停顿5秒钟
            Thread.sleep(5000L);

            // 发出10秒后的数据
            ctx.collect(new Event("Mary", "./home", 5000L));
            Thread.sleep(5000L);

            // 发出10秒+1ms后的数据
            ctx.collect(new Event("Alice", "./cart", 5001L));
            Thread.sleep(5000L);
        }

        @Override
        public void cancel() {
        }
    }


}
